with scan_detail as (
    select
        waybill_no ,scan_type ,scan_time,network_code,dt
    from(
            select
                waybill_no ,scan_type ,scan_time,network_code,dt,
                row_number() over(partition by waybill_no order by scan_time desc) as rk
            from
                (   --揽收
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from jms_dwd.dwd_tab_barscan_taking_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --入仓
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from jms_dwd.dwd_tab_barscan_warehousing_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --到件
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from jms_dwd.dwd_tab_barscan_arrival_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --发件
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from  jms_dwd.dwd_tab_barscan_send_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --派件
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from  jms_dwd.dwd_tab_barscan_deliver_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --入库 出库 快件取出
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from  jms_dwd.dwd_tab_barscan_collect_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --签收
                    select waybill_no ,scan_type ,signed_time as scan_time,signed_site as network_code,dt
                    from jms_dwd.dwd_tab_barscan_sign_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --建包扫描
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from jms_dwd.dwd_tab_barscan_bagging_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                    union all
                    --到解车扫描\拆包扫描\留仓件入仓\封发车扫描
                    select waybill_no ,scan_type ,scan_time,scan_site_code as network_code,dt
                    from jms_dwd.dwd_tab_barscan_other_base_hi
                    where dt = '{{ execution_date | cst_hour }}'
                ) tt
        )t  where rk=1
),
filter_bill as (
    select waybill_no
    from (
             --完结件
             select waybill_no
             from jms_dwd.dwd_tab_end_piece_base_hi
             where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -30) and '{{ execution_date | cst_ds }}'
             union all
             --问题件扫描
             select waybill_no
             from jms_dwd.dwd_tab_barscan_difficult_base_hi
             where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -30) and '{{ execution_date | cst_ds }}'
         )t group by waybill_no
),
order_info as (
    select waybill_id,order_source_code
    from jms_dwd.dwd_yl_oms_oms_order_hf
    where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
      and order_source_code='D09' --拼多多
),
abnormal_info as (
    select
        t1.network_code
         ,t1.network_type_desc
         ,t1.abnormal_causes
         ,t1.create_time
         ,t1.estimate_time
         ,t2.name --二级异常类型
         ,t3.one_level_express_status
         ,t3.two_level_express_status
    from
        (   select * from (
                              select network_code,
                                     network_type_desc,
                                     abnormal_causes,
                                     create_time,
                                     estimate_time,
                                     level_one_type,
                                     abnormal_type_id,
                                     row_number() over(partition by network_code order by update_time desc) as rk
                              from jms_dim.dim_secondary_abnormal_information_base_hi
                              where dt >= '2022-11-01'  --要用到全量数据
                                and create_by_name!='筛单'
                      and network_code is not null
                      and is_enable=1
                          )t where rk=1
        )t1
        inner join
        (
            select level_one_type,name,id
            from jms_dim.dim_secondary_abnormal_type_base
            group by level_one_type,name,id
        )t2 on t1.level_one_type=t2.level_one_type and t1.abnormal_type_id=t2.id
        inner join jms_tmp.tmp_abnormal_type_relat t3  on t2.name=t3.jt_abnormal_type

),
tmp as (
    select
        scan_detail.waybill_no
         ,scan_detail.scan_type
         ,scan_detail.scan_time
         ,scan_detail.network_code
         ,net.name as network_name
         ,ab.abnormal_causes --异常原因
         --,from_unixtime(unix_timestamp() ,'yyyy-MM-dd HH:mm:ss') as create_time --创建时间
         ,ab.create_time as create_time --创建时间
         ,ab.estimate_time  --预计解决时间 异常事件结束时间
         ,ab.network_type_desc  --事件来源
         ,ab.one_level_express_status as first_type   --一级事件类型
         ,ab.two_level_express_status as second_type  --二级事件类型
         ,ab.name as abnormal_name
         ,scan_detail.dt
    from scan_detail
         inner join order_info on scan_detail.waybill_no=order_info.waybill_id
         inner join abnormal_info ab  on ab.network_code=scan_detail.network_code
         left join filter_bill on scan_detail.waybill_no=filter_bill.waybill_no
         left join jms_dim.dim_network_whole_massage net on net.code=scan_detail.network_code
    where filter_bill.waybill_no is null            
)
insert overwrite table jms_dm.dm_sqs_abnormal_to_pdd_hi partition(dt)
select
     waybill_no as waybill_no
     ,scan_type
     ,scan_time
     ,network_code
     ,network_name
     ,abnormal_causes--异常原因
     ,create_time --创建时间
     ,estimate_time --预计解决时间 异常事件结束时间
     ,network_type_desc  --事件来源
     ,first_type  --一级事件类型
     ,second_type --二级事件类型
     ,dt
from tmp
distribute by (substr(scan_time,18,2))%10
